﻿<?php
/*
----------------------------------------------------------------------------------------------------------------
HTTP Simple Queue Service - httpsqs client class for PHP v1.0

Author: Zhang Junlei (zhangjunlei26@gmail.com)
----------------------------------------------------------------------------------------------------------------
Useage:
<?php
include_once("HttpSQS.class.php");
$httpsqs = new HttpSQS($host,$port,$charset);

$result = $httpsqs->put($name, $data); //1. PUT text message into a queue. If PUT successful, return boolean: true. If an error occurs, return boolean: false. If queue full, return text: HTTPSQS_PUT_END
$result = $httpsqs->get($name); //2. GET text message from a queue. Return the queue contents. If an error occurs, return boolean: false. If there is no unread queue message, return text: HTTPSQS_GET_END
$result = $httpsqs->gets($name); //3. GET text message and pos from a queue. Return example: array("pos" => 7, 'data' => "text message"). If an error occurs, return boolean: false. If there is no unread queue message, return: array("pos" => 0, 'data' => "HTTPSQS_GET_END")
$result = $httpsqs->status($name); //4. View queue status
$result = $httpsqs->status_json($name); //5. View queue status in json. Return example: {"name":"queue_name","maxqueue":5000000,"putpos":130,"putlap":1,"getpos":120,"getlap":1,"unread":10}
$result = $httpsqs->view($name, $pos); //6. View the contents of the specified queue pos (id). Return the contents of the specified queue pos.
$result = $httpsqs->reset($name); //7. Reset the queue. If reset successful, return boolean: true. If an error occurs, return boolean: false
$result = $httpsqs->maxqueue($name, $num); //8. Change the maximum queue length of per-queue. If change the maximum queue length successful, return boolean: true. If  it be cancelled, return boolean: false
$result = $httpsqs->synctime($name, $num); //9. Change the interval to sync updated contents to the disk. If change the interval successful, return boolean: true. If  it be cancelled, return boolean: false
----------------------------------------------------------------------------------------------------------------
*/
class HttpSQS
{

	protected $pconnect = TRUE;

	protected $host = 'localhost';

	protected $port = 1218;

	protected $charset = 'utf-8';

	protected $socket;

	public static $sockets = array();

	protected $errno;

	protected $errstr;

	public $content;

	public function __construct ($host = NULL ,$port = NULL ,$charset = NULL ,$pconnect = TRUE)
	{
		if ($host) $this->host = $host;
		if ($port) $this->port = $port;
		if ($charset) $this->charset = $charset;
		if (! $pconnect) $this->pconnect = FALSE;
	}

	protected function connect ()
	{
		if ($this->socket) return $this->socket;
		if ($this->pconnect)
		{
			$key = "{$this->host}:{$this->port}";
			if (isset(self::$sockets[$key]) && is_resource(self::$sockets[$key]))
			{
				return self::$sockets[$key];
			}
			self::$sockets[$key] = $this->socket = pfsockopen($this->host, $this->port, $this->errno, $this->errstr, 5);
		}
		else
		{
			$this->socket = fsockopen($this->host, $this->port, $this->errno, $this->errstr, 5);
		}
		return $this->socket;
	}

	protected function query ($request ,&$pos_value = 0)
	{
		$httpsqs_socket = $this->connect();
		if (! $httpsqs_socket) return FALSE;
		$header = array();
		fwrite($httpsqs_socket, $request);
		$line = trim(fgets($httpsqs_socket));
		$header[] = $line;
		list ( $proto, $rcode, $result ) = explode(" ", $line);
		$len = - 1;
		$pos_value = 0;
		while (($line = trim(fgets($httpsqs_socket))) != "")
		{
			$header[] = $line;
			if (strstr($line, "Content-Length:"))
			{
				list ( $cl, $len ) = explode(" ", $line);
			}
			if (strstr($line, "Pos:"))
			{
				list ( $pos_key, $pos_value ) = explode(" ", $line);
			}
		}
		if ($len < 0)
		{
			return false;
		}
		$body = fread($httpsqs_socket, $len);
		if (! $this->pconnect)
		{
			fclose($httpsqs_socket);
		}
		if ($body == 'HTTPSQS_ERROR' || $body == false)
		{
			$this->errstr = 'HTTPSQS_ERROR';
			return false;
		}
		return $body;
	}

	protected function http_get ($query ,&$pos_value = 0)
	{
		$connection = ($this->pconnect) ? 'Keep-Alive' : 'close';
		$request = "GET {$query} HTTP/1.1\r\nHost: {$this->host}\r\nConnection: {$connection}\r\n\r\n";
		return $this->query($request, $pos_value);
	}

	protected function http_post ($query ,$body)
	{
		$connection = ($this->pconnect) ? 'Keep-Alive' : 'close';
		if (! is_scalar($body))
		{
			$body = serialize($body);
		}
		$len = strlen($body);
		$request = "POST {$query} HTTP/1.1\r\nHost: {$this->host}\r\nContent-Length: {$len}\r\nConnection: {$connection}\r\n\r\n{$body}";
		return $this->query($request);
	}

	function put ($name ,$data ,$charset = NULL)
	{
		if (! $charset) $charset = $this->charset;
		$result = $this->http_post("/?charset={$charset}&name={$name}&opt=put", $data);
		if (! $result)
		{
			return FALSE;
		}
		elseif ($result == 'HTTPSQS_PUT_OK')
		{
			return true;
		}
		else if ($result == 'HTTPSQS_PUT_END')
		{
			return $result;
		}
		return false;
	}

	function get ($name ,$return_array = FALSE ,$charset = NULL)
	{
		if (! $charset) $charset = $this->charset;
		$pos = 0;
		$result = $this->http_get("/?charset={$charset}&name={$name}&opt=get", $pos);
		if($result==='HTTPSQS_GET_END')return NULL;
		if ($return_array && $result)
		{
			return array(
				'pos' => $pos, 
				'data' => $result
			);
		}
		return $result;
	}

	function status ($name ,$return_json = FALSE)
	{
		return $this->http_get("/?charset={$this->charset}&name={$name}&opt=status");
	}

	function view ($name ,$pos)
	{
		return $this->http_get("/?charset={$this->charset}&name={$name}&opt=view&pos={$pos}");
	}

	function reset ($name)
	{
		$result = $this->http_get("/?charset={$this->charset}&name={$name}&opt=reset");
		if ($result == 'HTTPSQS_RESET_OK') return true;
		return false;
	}

	function maxqueue ($name ,$num)
	{
		$result = $this->http_get("/?charset={$this->charset}&name={$name}&opt=maxqueue&num={$num}");
		if ($result == 'HTTPSQS_MAXQUEUE_OK') return true;
		return false;
	}

	function status_json ($name)
	{
		return $this->http_get("/?charset={$this->charset}&name={$name}&opt=status_json");
	}

	function synctime ($name ,$num)
	{
		$result = $this->http_get("/?charset={$this->charset}&name={$name}&opt=synctime&num={$num}");
		if ($result == 'HTTPSQS_SYNCTIME_OK') return true;
		return false;
	}
}
